/**
 * Copyright (c) 2022 Presto Labs Pte. Ltd.
 * Author: Ju
 */

#include "coin2/controller/controller.h"

#include <glog/logging.h>
#include <sys/types.h>
#include <unistd.h>

#include <cstdlib>
#include <ctime>

#include "coin2/base/time.h"

namespace {
static const uint32_t Port = 9000;

std::string GetPath(const std::string& method) {
  int32_t sz = method.size();
  for (int32_t i = 0; i < sz; ++i) {
    if (method[i] == '/') {
      return method.substr(i + 1);
    }
  }
  return "";
}

}  // namespace
namespace coin2::controller {

Controller::Controller(boost::asio::io_context* ioc, std::string strategy_name)
    : ioc_(ioc),
      strategy_name_(strategy_name),
      tcp_client_(new TcpClient(ioc_, this)),
      timer_(*ioc_) {
  srand((unsigned)time(NULL));
  node_id_ = std::rand();
  char local_host_name[256];
  if (gethostname(local_host_name, 256)) {
    LOG(ERROR) << "Get Host Name Error!";
    host_name_ = "default";
  } else {
    host_name_ = local_host_name;
  }
  pid_ = getpid();
  Connect();
}

Controller::~Controller() {}

void Controller::Connect() {
  tcp_client_->Connect("127.0.0.1", Port);
  OnTimer();
}

bool Controller::Find(const std::string& str) {
  auto iter = tasks.find(str);
  return iter != tasks.end();
}

void Controller::ExportMethods(const MethodsVector& v) {
  for (auto& it : v) {
    auto sys_name = it.first;
    for (auto& methods : it.second) {
      Register(it.first + "/" + methods.first, methods.second);
    }
  }
}

void Controller::Execute(uint32_t dst_id, nlohmann::json payload) {
  auto path = GetPath(payload["Method"]);
  if (!Find(path)) {
    LOG(WARNING) << "No found func_path!";
    payload["Result"] = "No Found End Point";
    tcp_client_->SendMsg(MakeResponse(node_id_, dst_id, std::move(payload)));
    return;
  }

  auto task = tasks.at(path);
  task(dst_id, std::move(payload));
}

void Controller::OnProcessMessage(memory::MemSlice buffer) {
  boost::asio::post(*ioc_, [this, data = std::move(buffer)]() {
    base::Unpacker unpk(data.Begin(), data.GetUsedSize(), true);
    Packet pkt;
    unpk >> pkt;
    nlohmann::json payload;
    if (pkt.encoding == MessageType::Json) {
      payload = nlohmann::json::parse(pkt.payload.begin(), pkt.payload.end());
    } else {
      payload = nlohmann::json::from_msgpack(pkt.payload);
    }
    if (pkt.type == PayloadType::Heartbeat) {
      LOG(INFO) << " node_id: " << payload["Info"]["NodeId"]
                << " node_name: " << payload["Info"]["NodeName"]
                << " host_name: " << payload["Info"]["Hostname"]
                << " last_seen: " << payload["Info"]["LastSeen"];
      return;
    }
    Execute(pkt.src_id, payload);
  });
}
void Controller::OnBroken() { tcp_client_->RetryConnect(); }

void Controller::OnTimer() {
  timer_.expires_from_now(boost::posix_time::seconds(15));
  timer_.async_wait([this](const boost::system::error_code& ec) {
    tcp_client_->SendMsg(SendHeartbeat(node_id_, strategy_name_, host_name_,
                                       pid_, impl::GetCurrentTimestamp(), false,
                                       ++req_num_));
    OnTimer();
  });
}

}  // namespace coin2::controller